package t20240513;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class RedisSinkTest2 {


    public static void main(String[] args) {

        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(2);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);


        String sourceSql = "CREATE TABLE datagen (" +
                "    score INT," +
                "    sn STRING" +
                "  ) WITH (" +
                "    'connector' = 'datagen'," +
                "    'rows-per-second'='1'," +
                "    'fields.score.kind'='sequence'," +
                "    'fields.score.start'='1'," +
                "    'fields.score.end'='10'," +
                "    'fields.sn.length'='4'" +
                "  )";

        String sinkSql =
                "  CREATE TABLE redis_sink (" +
                        "    key STRING," +
                        "    score INT," +
                        "    sn STRING," +
                        "    mins STRING," +
                        "    maxs STRING," +
                        "    primary key (key) not ENFORCED" +
                        "    ) WITH (" +
                        "    'connector' = 'redis'," +
                        "    'host' = '127.0.0.1'," +
                        "    'port' = '6379'," +
                        "    'redis-mode' = 'single'," +
                        "    'timeout' = '2000'," +
                        "    'password' = 'yy2024'," +
                        "    'command' = 'ZADD'," +
                        "    'zset.zremrangeby' = 'LEX'," +
                        "    'max.retries' = '5'" +
                        "  )";

        String execSql = "INSERT INTO redis_sink select 'SN_TEST' as key,100 as score,CONCAT('aa', sn) sn,'a' as mins ,'c' as maxs from datagen";
//        String execSql = "INSERT INTO print_table select stime,sn,UNIX_TIMESTAMP(DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy-MM-dd hh:mm:ss'), 'yyyy-MM-dd hh:mm:ss') as updatetime from datagen";

        tableEnvironment.executeSql(sourceSql);
        System.out.println("-----1");
        tableEnvironment.executeSql(sinkSql);
        System.out.println("-----2");
        tableEnvironment.executeSql(execSql);

    }
}
